﻿/* Copyright (C) 2007-2010 STS Soft

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
using STSdb.General.Collections;
using STSdb.General.Buffers;
using STSdb.Files;
using STSdb.Persist;

namespace STSdb.Data
{
    /// <summary>
    ///Each row in table is represented by a (key; record) pair. The key is unique per row.
    ///
    ///Table provides fast operations by key: adding, replacing, searching, deleting, iterating, etc. 
    ///
    ///All changes in the table are encapsulated by a transaction, so they can be committed and rollbacked. 
    ///It is possible to request a snapshot at any time. In this case any further changes to table do not affect the snapshot.
    ///Time consumed by Commit, Rollback and Snapshot is constant, it is not affected by the size of transaction.
    /// </summary>
    /// <typeparam name="TKey">Type of keys.</typeparam>
    /// <typeparam name="TRecord">Type of records.</typeparam>
    /// <example> Getting started.
    /// <code lang="cs">
    /// <![CDATA[
    ///    using System;
    ///    using System.Collections.Generic;
    ///    using System.Linq;
    ///    using System.Text;
    ///    using Repository.Data;
    ///
    ///    namespace GettingStarted
    ///    {
    ///        /// <summary>
    ///        /// Record type for examples
    ///        /// </summary>
    ///        public class Tick
    ///        {
    ///            public string Symbol { get; set; }
    ///            public DateTime Timestamp { get; set; }
    ///            public double Bid { get; set; }
    ///            public double Ask { get; set; }
    ///
    ///            public Tick()
    ///            {
    ///            }
    ///        }
    ///        
    ///        public class GettingStarted
    ///        {
    ///            public StorageEngine Repository { get; private set; }
    ///            public string FileName { get; private set; }
    ///
    ///            public GettingStarted(string fileName)
    ///            {
    ///                FileName = fileName;
    ///            }
    ///
    ///            /// <summary>
    ///            /// 0. Open database
    ///            /// </summary>
    ///            public void OpenRepository()
    ///            {
    ///                //Open database
    ///                Repository = StorageEngine.FromFile(FileName);
    ///            }
    ///
    ///            /// <summary>
    ///            /// 1. Add ticks
    ///            /// </summary>
    ///            public ulong Add(string tableName, int count)
    ///            {
    ///                //open table
    ///                XTable<ulong, Tick> table = Repository.Scheme.CreateOrOpenXTable<ulong, Tick>(new Locator(tableName));
    ///
    ///                //insert ticks
    ///                ulong key = table.Count > 0 ? table.LastRow.Key + 1 : 0;
    ///                foreach (Tick tick in RandomTicks(count))
    ///                {
    ///                    table[key] = tick;
    ///                    key++;
    ///                }
    ///
    ///                table.Commit();
    ///
    ///                //close table
    ///                table.Close();
    ///
    ///                return table.Count;
    ///            }
    ///
    ///            /// <summary>
    ///            /// 2. Read ticks
    ///            /// </summary>
    ///            public ulong Read(string tableName)
    ///            {
    ///                //create table
    ///                XTable<ulong, Tick> table = Repository.Scheme.CreateOrOpenXTable<ulong, Tick>(new Locator(tableName));
    ///
    ///                //read in ascending order
    ///                foreach (Row<ulong, Tick> row in table.Forward())
    ///                {
    ///                }
    ///
    ///                //close table
    ///                table.Close();
    ///
    ///                return table.Count;
    ///            }
    ///
    ///            /// <summary>
    ///            /// 3. Index by Timestamp
    ///            /// </summary>
    ///            public ulong IndexByTimestamp(string srcTableName, string dstTableName)
    ///            {
    ///                //open source table
    ///                XTable<ulong, Tick> srcTable = Repository.Scheme.CreateOrOpenXTable<ulong, Tick>(new Locator(srcTableName));
    ///
    ///                //recreate index table
    ///                Repository.Scheme.Delete(new Locator(dstTableName));
    ///                Repository.Scheme.Commit();
    ///                XTable<XKey<DateTime, ulong>, ulong> dstTable = Repository.Scheme.CreateOrOpenXTable<XKey<DateTime, ulong>, ulong>(new Locator(dstTableName));
    ///
    ///                ulong subkey = 0;
    ///                foreach (Row<ulong, Tick> row in srcTable)
    ///                {
    ///                    XKey<DateTime, ulong> key = new XKey<DateTime, ulong>(row.Record.Timestamp, subkey);
    ///                    dstTable[key] = row.Key;
    ///                    subkey++;
    ///                }
    ///
    ///                //close tables
    ///                srcTable.Close();
    ///                dstTable.Commit();
    ///                dstTable.Close();
    ///
    ///                return dstTable.Count;
    ///            }
    ///
    ///            /// <summary>
    ///            /// 3.1. Read through index
    ///            /// </summary>
    ///            public ulong ReadThroughIndex(string tableName, string indexTableName)
    ///            {
    ///                //open table
    ///                XTable<ulong, Tick> table = Repository.Scheme.CreateOrOpenXTable<ulong, Tick>(new Locator(tableName));
    ///
    ///                //open index table
    ///                XTable<XKey<DateTime, ulong>, ulong> indexTable = Repository.Scheme.CreateOrOpenXTable<XKey<DateTime, ulong>, ulong>(new Locator(indexTableName));
    ///
    ///                //read trough indexTable
    ///                foreach (var indexRow in indexTable.Forward())
    ///                {
    ///                    Tick record = table[indexRow.Record];
    ///                }
    ///
    ///                //close tables
    ///                table.Close();
    ///                indexTable.Commit();
    ///                indexTable.Close();
    ///
    ///                return indexTable.Count;
    ///            }
    ///
    ///            /// <summary>
    ///            /// 4. Sort by Timestamp
    ///            /// </summary>
    ///            public ulong SortByTimestamp(string srcTableName, string dstTableName)
    ///            {
    ///                //open source table
    ///                XTable<ulong, Tick> srcTable = Repository.Scheme.CreateOrOpenXTable<ulong, Tick>(new Locator(srcTableName));
    ///
    ///                //recreate destination table
    ///                Repository.Scheme.Delete(new Locator(dstTableName));
    ///                Repository.Scheme.Commit();
    ///
    ///                /* Sorting by DateTime is elegant - just define destination table with appropriate type of TKey and copy ticks into it.
    ///                 * TKey of destination table is usual a kind of XKey - a special predefined set of composite keys. In our case we have to 
    ///                 * use XKey with two components: DateTime and ulong. DateTime subkey is left (oldest) part of composite key, 
    ///                 * ulong subkey is always the rightest (youngest) part of composite key.
    ///                 */
    ///                XTable<XKey<DateTime, ulong>, Tick> dstTable = Repository.Scheme.CreateOrOpenXTable<XKey<DateTime, ulong>, Tick>(new Locator(dstTableName));
    ///
    ///                /*
    ///                 * In general all subkeys of composite key have to define maxLengths (in bytes). For most primitive types
    ///                 * (like (S)Byte, Boolean, (U)Int16, (U)Int32, (U)Int64, Single, Double, DateTime) these lengths are already internally defined. 
    ///                 * But for subkeys with non-fixed lengths (like byte[], string or custom subkeys) we have to specify MaxLength for each subkey type.
    ///                 */
    ///                //dstTable.KeyMap = new XKeyMap<DateTime, ulong>(sizeof(long), sizeof(ulong));
    ///
    ///                //sort algorithm
    ///                ulong subkey = 0;
    ///                foreach (Row<ulong, Tick> row in srcTable)
    ///                {
    ///                    XKey<DateTime, ulong> key = new XKey<DateTime, ulong>(row.Record.Timestamp, subkey);
    ///                    dstTable[key] = row.Record;
    ///                    subkey++;
    ///                }
    ///
    ///                //close tables
    ///                srcTable.Close();
    ///                dstTable.Commit();
    ///                dstTable.Close();
    ///
    ///                return dstTable.Count;
    ///            }
    ///
    ///            /// <summary>
    ///            /// 5. Split by Symbol
    ///            /// </summary>
    ///            public ulong SplitBySymbol(string srcTableName, string dstPathName)
    ///            {
    ///                //open source table
    ///                XTable<ulong, Tick> srcTable = Repository.Scheme.CreateOrOpenXTable<ulong, Tick>(new Locator(srcTableName));
    ///
    ///                Dictionary<string, XTable<ulong, Tick>> map = new Dictionary<string, XTable<ulong, Tick>>();
    ///                foreach (Row<ulong, Tick> row in srcTable)
    ///                {
    ///                    string symbol = row.Record.Symbol;
    ///
    ///                    XTable<ulong, Tick> dstTable;
    ///                    if (!map.TryGetValue(symbol, out dstTable))
    ///                        map[symbol] = dstTable = Repository.Scheme.CreateOrOpenXTable<ulong, Tick>(new Locator(dstPathName + symbol));
    ///
    ///                    dstTable[dstTable.Count] = row.Record;
    ///                }
    ///
    ///                //close source table
    ///                srcTable.Close();
    ///
    ///                //commit & close split tables
    ///                Transaction transaction = new Transaction(Repository, map.Values.ToArray());
    ///                transaction.Commit();
    ///                foreach (var kv in map)
    ///                    kv.Value.Close();
    ///
    ///                return (ulong)map.Count;
    ///            }
    ///
    ///            /// <summary>
    ///            /// 6. Update Ask
    ///            /// </summary>
    ///            public ulong UpdateAsk(string tableName)
    ///            {
    ///                //open source table
    ///                XTable<ulong, Tick> table = Repository.Scheme.CreateOrOpenXTable<ulong, Tick>(new Locator(tableName));
    ///
    ///                //take a snapshot of the table
    ///                XTable<ulong, Tick> snapshot = table.Snapshot();
    ///
    ///                foreach (var row in snapshot)
    ///                {
    ///                    //set fixed spread.
    ///                    row.Record.Ask = row.Record.Bid + 4 * 0.0001;
    ///
    ///                    //update record in table
    ///                    table[row.Key] = row.Record;
    ///                }
    ///
    ///                //commit & close
    ///                table.Commit();
    ///                table.Close();
    ///
    ///                //It is not necessary to close or commit the snapshot,
    ///                //snapshot.Close();
    ///                //snapshot.Commit();
    ///
    ///                //But if you want to keep and develop table snapshot as a separate branch, then register it in StorageEngine's Scheme.
    ///                //Repository.Register(snapshot, newLocator);
    ///
    ///                return table.Count;
    ///            }
    ///
    ///            /// <summary>
    ///            /// 7. Close database
    ///            /// </summary>
    ///            public void CloseRepository()
    ///            {
    ///                //commit changes to scheme
    ///                Repository.Scheme.Commit();
    ///                Repository.Scheme.Close();
    ///
    ///                //Close database
    ///                Repository.Dispose();
    ///            }
    ///
    ///            /// <summary>
    ///            /// Generate random ticks
    ///            /// </summary>
    ///            private static IEnumerable<Tick> RandomTicks(int count)
    ///            {
    ///                Random random = new Random();
    ///
    ///                //Some symbols
    ///                string[] symbols = new string[] { "EURUSD", "GBPCHF", "EURGBP", "JPYUSD", "GBPCAD" };
    ///                double[] prices = new double[symbols.Length];
    ///                double[] pipsizes = new double[symbols.Length];
    ///                int[] digits = new int[symbols.Length];
    ///
    ///                //Initialize startup data
    ///                for (int i = 0; i < symbols.Length; i++)
    ///                {
    ///                    prices[i] = 1.5;
    ///                    pipsizes[i] = 0.0001;
    ///                    digits[i] = 4;
    ///                }
    ///
    ///                DateTime baseTime = DateTime.Now;
    ///
    ///                //Generate ticks
    ///                for (int i = 0; i < count; i++)
    ///                {
    ///                    int id = random.Next(symbols.Length);
    ///
    ///                    //generate random movement
    ///                    int pips = random.Next(0, 10);
    ///                    int direction = random.Next() % 2 == 0 ? 1 : -1;
    ///                    int spread = random.Next(2, 30);
    ///                    int seconds = random.Next(0, 5 * 365 * 24 * 60 * 60);
    ///
    ///                    //generate values
    ///                    string symbol = symbols[id];
    ///
    ///                    DateTime timestamp = baseTime.AddSeconds(-seconds);
    ///
    ///                    double bid = prices[id];
    ///                    bid += direction * pips * pipsizes[id];
    ///                    bid = Math.Round(bid, digits[id]);
    ///
    ///                    double ask = bid + spread * pipsizes[id];
    ///                    ask = Math.Round(ask, digits[id]);
    ///
    ///                    //create tick
    ///                    Tick tick = new Tick();
    ///                    tick.Symbol = symbol;
    ///                    tick.Timestamp = timestamp;
    ///                    tick.Bid = bid;
    ///                    tick.Ask = ask;
    ///
    ///                    yield return tick;
    ///                }
    ///            }
    ///        }
    ///
    ///    }
    ///]]>
    /// </code>
    /// </example>
    public class XTable<TKey, TRecord> : ITransaction, IEnumerable<Row<TKey, TRecord>>
    {
        private const ulong HALF_HEADER_SIZE = 32;
        internal const ulong HEADER_SIZE = sizeof(byte) + 2 * HALF_HEADER_SIZE;

        private byte ActiveHeader;
        private FTable<TKey, TRecord>[] Cache = new FTable<TKey, TRecord>[8];
        private LinkedList<FTable<TKey, TRecord>.Link> SharedCache = new LinkedList<FTable<TKey, TRecord>.Link>();
        private FTable<TKey, TRecord> ServiceTable;

        //Depth -> Offset
        private FTable<uint, ulong> Scheme;
        private byte[] MinIndex = new byte[] { 0x00 };
        private byte[] MaxIndex = new byte[] { 0xFF };

        /// <summary>
        /// Container file.
        /// </summary>
        public readonly RawFile RawFile;
        /// <summary>
        /// Offset in RawFile.
        /// </summary>
        public readonly ulong Offset;

        /// <summary>
        /// 
        /// </summary>
        /// <param name="rawFile">Container file obtained from FileSystem class.</param>
        /// <param name="offset">Offset in rawFile.</param>
        public XTable(RawFile rawFile, ulong offset)
        {
            if (rawFile == null)
                throw new ArgumentNullException("rawFile");

            Offset = offset;
            RawFile = rawFile;
            BinaryFile binaryFile = new BinaryFile(RawFile);

            //FTable.Offset -> FTable.ActiveHeader
            FTable<ulong, byte> journalTable = new FTable<ulong, byte>(RawFile, Offset + HEADER_SIZE);
            journalTable.Depth = sizeof(ulong);

            foreach (var row in journalTable)
            {
                binaryFile.Position = row.Key;
                binaryFile.Write(row.Record);
            }
            journalTable.Clear();
            journalTable.Commit();

            binaryFile.Position = Offset;
            ActiveHeader = binaryFile.ReadByte();

            LoadHeader(ActiveHeader);
            switch (Version) //it's a new file
            {
                case 0:
                    {
                        Version = 1;
                        SaveHeader(ActiveHeader);
                        binaryFile.Position = Offset + unchecked(HEADER_SIZE - 1);
                        binaryFile.Write((byte)0); //marker for end of the header
                        RawFile.Flush();
                    }
                    break;
                case 1:
                    {
                    }
                    break;
                default:
                    throw new Exception(String.Format("Unknown version {0}", Version));
            }

            Scheme = new FTable<uint, ulong>(RawFile, journalTable.Offset + FTable<ulong, byte>.HEADER_SIZE);
            Scheme.Depth = sizeof(uint);

            ServiceTable = new FTable<TKey, TRecord>(RawFile, Scheme.Offset + FTable<uint, ulong>.HEADER_SIZE);
        }

        #region Manage

        /// <summary>
        /// Gets or sets the record associated with the specified key.
        /// </summary>
        /// <param name="key">The key of the record to get or set.</param>
        /// <returns>The record associated with the specified key. If the specified key is not found, a get operation throws a KeyNotFoundException, and a set operation creates a new record with the specified key.</returns>
        /// <exception cref="KeyNotFoundException">The property is retrieved and key does not exist in the collection.</exception>
        public TRecord this[TKey key]
        {
            get
            {
                byte[] index = KeyMap.Direct(key);

                FTable<TKey, TRecord> table = Retrieve((uint)index.Length, false);
                if (table == null)
                    throw new KeyNotFoundException(key.ToString());

                return table[index];
            }
            set
            {
                byte[] index = KeyMap.Direct(key);

                FTable<TKey, TRecord> table = Retrieve((uint)index.Length, true);
                ulong oldCount = table.Count;
                table[index] = value;
                Count += table.Count - oldCount;
            }
        }

        /// <summary>
        /// Deletes a record with the specified key.
        /// </summary>
        /// <param name="key">The key of the record to delete.</param>
        public void Delete(TKey key)
        {
            byte[] index = KeyMap.Direct(key);

            FTable<TKey, TRecord> table = Retrieve((uint)index.Length, false);
            if (table == null)
                return;

            ulong oldCount = table.Count;
            table.Delete(index);
            Count += table.Count - oldCount;

            if (table.Count == 0)
            {
                Cache[table.Depth] = null;
                Scheme.Delete(table.Depth);
            }
        }

        /// <summary>
        /// Deletes all records with keys in range [firstKey, lastKey].
        /// </summary>
        /// <param name="firstKey">The first key of the range of records to delete.</param>
        /// <param name="lastKey">The last key of the range of records to delete.</param>
        /// <exception cref="ArgumentException">If firstKey is greater than lastKey</exception>
        public void Delete(TKey firstKey, TKey lastKey)
        {
            byte[] firstIndex = KeyMap.Direct(firstKey);
            byte[] lastIndex = KeyMap.Direct(lastKey);

            if (firstIndex.Length > lastIndex.Length)
                return;

            ulong oldCount;
            FTable<TKey, TRecord> table;
            if (firstIndex.Length == lastIndex.Length)
            {
                table = Retrieve((uint)firstIndex.Length, false);
                if (table != null)
                {
                    oldCount = table.Count;
                    table.Delete(firstIndex, lastIndex);
                    Count += table.Count - oldCount;
                }
            }
            else
            {
                table = Retrieve((uint)firstIndex.Length, false);
                if (table != null)
                {
                    oldCount = table.Count;
                    table.Delete(firstIndex, table.MaxIndex);
                    Count += table.Count - oldCount;
                }

                table = Retrieve((uint)lastIndex.Length, false);
                if (table != null)
                {
                    oldCount = table.Count;
                    table.Delete(table.MinIndex, lastIndex);
                    Count += table.Count - oldCount;
                }

                uint firstDepth = (uint)firstIndex.Length + 1;
                uint lastDepth = (uint)lastIndex.Length - 1;
                foreach (var row in Scheme.Forward(firstDepth, lastDepth))
                {
                    uint depth = row.Key;
                    table = Retrieve(depth, false);

                    Count -= table.Count;
                    Cache[table.Depth] = null;
                }
                Scheme.Delete(firstDepth, lastDepth);
            }
        }

        /// <summary>
        /// Deletes all records from the table.
        /// </summary>
        public void Clear()
        {
            Cache = new FTable<TKey, TRecord>[Cache.Length];
            Scheme.Clear();
            SharedCache.Clear();

            Count = 0;
        }

        /// <summary>
        /// Commit changes and close the table.
        /// </summary>
        public void Close()
        {
            foreach (var table in Cache)
                if (table != null)
                    table.Close();
        }

        /// <summary>
        /// Adds all rows from table. On key conflict rows from specified table replace avaiable rows.
        /// </summary>
        /// <param name="table"></param>
        public void Merge(XTable<TKey, TRecord> table)
        {
            foreach (var row in table)
                this[row.Key] = row.Record;
        }

        public void Change(TKey key, TKey newKey)
        {
            TRecord record = this[key];
            Delete(key);
            this[newKey] = record;
        }

        public void Swap(TKey key1, TKey key2)
        {
            TRecord record1 = this[key1];
            TRecord record2 = this[key2];
            this[key1] = record2;
            this[key2] = record1;
        }

        #endregion

        #region Find Methods

        /// <summary>
        /// Gets the record associated with the specified key.
        /// </summary>
        /// <param name="key">The key of the record to get.</param>
        /// <param name="record">When this method returns, contains the record associated with the specified key, if the key is found; otherwise, the default value for the type of the value parameter. This parameter is passed uninitialized.</param>
        /// <returns>true if the XTable<TKey, TValue> contains a record with the specified key; otherwise, false.</returns>
        public bool TryGet(TKey key, out TRecord record)
        {
            record = default(TRecord);

            byte[] index = KeyMap.Direct(key);

            FTable<TKey, TRecord> table = Retrieve((uint)index.Length, false);
            if (table == null)
                return false;

            return table.TryGet(index, out record);
        }

        public TRecord TryGetDefault(TKey key, TRecord defaultRecord)
        {
            TRecord record;
            if (TryGet(key, out record))
                return record;

            return defaultRecord;
        }

        /// <summary>
        /// Searches for a row with the specified key.
        /// </summary>
        /// <param name="key">The key of the row to get.</param>
        /// <returns>Row with the specified key or null, if a row with such key does not exists.</returns>
        public TRecord Find(TKey key)
        {
            TRecord record;
            TryGet(key, out record);

            return record;
        }

        /// <summary>
        /// Determines whether the table contains a record with the specified key.
        /// </summary>
        /// <param name="key">The key to locate in the table.</param>
        /// <returns>True, if the table contains an element with the specified key; otherwise, false.</returns>
        public bool Exists(TKey key)
        {
            TRecord record;
            return TryGet(key, out record);
        }

        internal Row<TKey, TRecord> FindNext(byte[] index)
        {
            return Forward(index, MaxIndex).FirstOrDefault();
        }

        /// <summary>
        /// Searches for a row with key greater than or equals to the specified key.
        /// </summary>
        /// <param name="key">The key of the row to get.</param>
        /// <returns>If the key is found it returns the record associated with it. If the key
        /// doesn't exist it returns the first record with a key greater than the
        /// specified one. If there are no records with keys greater or equal to
        /// the specified one it returns null.</returns>
        public Row<TKey, TRecord> FindNext(TKey key)
        {
            byte[] index = KeyMap.Direct(key);
            return FindNext(index);
        }

        internal Row<TKey, TRecord> FindPrev(byte[] index)
        {
            return Backward(index, MinIndex).FirstOrDefault();
        }

        /// <summary>
        /// Searches for a row with key less than or equals to the specified key.
        /// </summary>
        /// <param name="key">The key of the row to get.</param>
        /// <returns>If the key is found it returns the record associated with it. If the key
        /// doesn't exist it returns the first record with a key less than the
        /// specified one. If there are no records with keys less or equal to
        /// the specified one it returns null.</returns>
        public Row<TKey, TRecord> FindPrev(TKey key)
        {
            byte[] index = KeyMap.Direct(key);
            return FindPrev(index);
        }

        /// <summary>
        /// Searches for a row with key greater than the specified.
        /// </summary>
        /// <param name="key">The key of the row to get.</param>
        /// <returns>The first record with a key greater than the specified one.
        /// If there are no records with keys greater than the specified one it returns null.</returns>
        public Row<TKey, TRecord> FindAfter(TKey key)
        {
            byte[] index = KeyMap.Direct(key);
            if (index.Increment())
            {
                if (index.Length >= MaxIndex.Length)
                    return null;

                byte[] idx = new byte[index.Length + 1];
                //for (int i = 0; i < idx.Length; i++)
                //idx[i] = byte.MinValue;
                return FindNext(idx);
            }

            return FindNext(index);
        }

        /// <summary>
        /// Searches for a row with key less than the specified.
        /// </summary>
        /// <param name="key">The key of the row to get.</param>
        /// <returns>The first record with a key less than the specified one.
        /// If there are no records with keys less than the specified one it returns null.</returns>
        public Row<TKey, TRecord> FindBefore(TKey key)
        {
            byte[] index = KeyMap.Direct(key);
            if (index.Decrement())
            {
                if (index.Length <= MinIndex.Length)
                    return null;

                byte[] idx = new byte[index.Length - 1];
                for (int i = 0; i < idx.Length; i++)
                    idx[i] = byte.MaxValue;
                return FindPrev(idx);
            }

            return FindPrev(index);
        }

        #endregion

        #region Enumerators

        public IEnumerable<Row<TKey, TRecord>> Forward(byte[] firstIndex, byte[] lastIndex)
        {
            byte[] fromIndex, toIndex;
            foreach (var row in Scheme.Forward((uint)firstIndex.Length, (uint)lastIndex.Length))
            {
                uint depth = row.Key;
                FTable<TKey, TRecord> table = Retrieve(depth, false);

                if (depth == firstIndex.Length)
                    fromIndex = firstIndex;
                else
                    fromIndex = table.MinIndex;

                if (depth == lastIndex.Length)
                    toIndex = lastIndex;
                else
                    toIndex = table.MaxIndex;

                foreach (var innerRow in table.Forward(fromIndex, toIndex))
                    yield return innerRow;
            }
        }

        public IEnumerable<Row<TKey, TRecord>> Forward(byte[] firstIndex)
        {
            return Forward(firstIndex, MaxIndex);
        }

        public IEnumerable<Row<TKey, TRecord>> Backward(byte[] lastIndex, byte[] firstIndex)
        {
            byte[] fromIndex, toIndex;
            foreach (var row in Scheme.Backward((uint)lastIndex.Length, (uint)firstIndex.Length))
            {
                uint depth = row.Key;
                FTable<TKey, TRecord> table = Retrieve(depth, false);

                if (depth == firstIndex.Length)
                    fromIndex = firstIndex;
                else
                    fromIndex = table.MinIndex;

                if (depth == lastIndex.Length)
                    toIndex = lastIndex;
                else
                    toIndex = table.MaxIndex;

                foreach (var innerRow in table.Backward(toIndex, fromIndex))
                    yield return innerRow;
            }
        }


        public IEnumerable<Row<TKey, TRecord>> Backward(byte[] lastIndex)
        {
            return Forward(lastIndex, MinIndex);
        }

        /// <summary>
        /// Enumerates all records with keys in range [firstKey, lastKey] in ascending order.
        /// </summary>
        /// <param name="firstKey">left border of the range.</param>
        /// <param name="lastKey">right border of the range.</param>
        /// <returns>Row enumerator.</returns>
        public IEnumerable<Row<TKey, TRecord>> Forward(TKey firstKey, TKey lastKey)
        {
            byte[] firstIndex = KeyMap.Direct(firstKey);
            byte[] lastIndex = KeyMap.Direct(lastKey);

            return Forward(firstIndex, lastIndex);
        }

        /// <summary>
        /// Enumerates all records with keys in range [firstKey, lastKey] in descending order.
        /// </summary>
        /// <param name="lastKey">right border of the range.</param>
        /// <param name="firstKey">left border of the range.</param>
        /// <returns>Row enumerator.</returns>
        public IEnumerable<Row<TKey, TRecord>> Backward(TKey lastKey, TKey firstKey)
        {
            byte[] firstIndex = KeyMap.Direct(firstKey);
            byte[] lastIndex = KeyMap.Direct(lastKey);

            return Backward(lastIndex, firstIndex);
        }

        /// <summary>
        /// Enumerates all records with keys greater than or equal to the specified key in ascending order.
        /// </summary>
        /// <param name="firstKey">enumeration initial key.</param>
        /// <returns>Row enumerator.</returns>
        public IEnumerable<Row<TKey, TRecord>> Forward(TKey firstKey)
        {
            byte[] firstIndex = KeyMap.Direct(firstKey);
            return Forward(firstIndex, MaxIndex);
        }

        /// <summary>
        /// Enumerates all records with keys less than or equal to the specified key in descending order.
        /// </summary>
        /// <param name="lastKey">enumeration initial key.</param>
        /// <returns>Row enumerator.</returns>
        public IEnumerable<Row<TKey, TRecord>> Backward(TKey lastKey)
        {
            byte[] lastIndex = KeyMap.Direct(lastKey);
            return Backward(lastIndex, MinIndex);
        }

        /// <summary>
        /// Enumerates all records in descending order.
        /// </summary>
        /// <returns>Row enumerator.</returns>
        public IEnumerable<Row<TKey, TRecord>> Backward()
        {
            return Backward(MaxIndex, MinIndex);
        }

        /// <summary>
        /// Enumerates all records in ascending order.
        /// </summary>
        /// <returns>Row enumerator.</returns>
        public IEnumerable<Row<TKey, TRecord>> Forward()
        {
            return Forward(MinIndex, MaxIndex);
        }

        public IEnumerable<Row<TKey, TRecord>> ForwardGroup(byte[] index)
        {
            if (index == null)
                throw new ArgumentNullException("index");

            foreach (var row in Scheme.Forward((uint)index.Length))
            {
                uint depth = row.Key;
                FTable<TKey, TRecord> table = Retrieve(depth, false);

                foreach (var innerRow in table.ForwardGroup(index))
                    yield return innerRow;
            }
        }

        public IEnumerable<Row<TKey, TRecord>> BackwardGroup(byte[] index)
        {
            if (index == null)
                throw new ArgumentNullException("index");

            foreach (var row in Scheme.Backward((uint)index.Length))
            {
                uint depth = row.Key;
                FTable<TKey, TRecord> table = Retrieve(depth, false);

                foreach (var innerRow in table.BackwardGroup(index))
                    yield return innerRow;
            }
        }

        public IEnumerable<Row<TKey, TRecord>> ForwardGroup(TKey key)
        {
            byte[] index = KeyMap.Direct(key);
            return ForwardGroup(index);
        }

        public IEnumerable<Row<TKey, TRecord>> BackwardGroup(TKey key)
        {
            byte[] index = KeyMap.Direct(key);
            return Backward(index, index);
        }

        #region IEnumerable<Row<TKey,TRecord>> Members

        public IEnumerator<Row<TKey, TRecord>> GetEnumerator()
        {
            foreach (var row in Scheme.Forward())
            {
                uint depth = row.Key;
                FTable<TKey, TRecord> table = Retrieve(depth, false);

                foreach (var innerRow in table)
                    yield return innerRow;
            }
        }

        #endregion

        #region IEnumerable Members

        IEnumerator IEnumerable.GetEnumerator()
        {
            return GetEnumerator();
        }

        #endregion

        /// <summary>
        /// Make an isolated copy (snapshot) of the table. 
        /// </summary>
        /// <returns>Snapshot of the table.</returns>
        public XTable<TKey, TRecord> Snapshot()
        {
            XTable<TKey, TRecord> table = new XTable<TKey, TRecord>(RawFile, RawFile.Length);
            table.MinIndex = MinIndex;
            table.MaxIndex = MaxIndex;
            table.RecordPersist = RecordPersist;
            table.KeyMap = KeyMap;
            table.Count = Count; //don't move it above the two lines.
            table.CacheCapacity = CacheCapacity;
            table.BlockCapacity = BlockCapacity;
            foreach (var row in Scheme.Forward())
            {
                FTable<TKey, TRecord> innerTable = Retrieve(row.Key, false);
                FTable<TKey, TRecord> snapshot = innerTable.Snapshot();
                snapshot.Cache = table.SharedCache;
                table.Packet(snapshot.Depth, snapshot);
                table.Scheme[snapshot.Depth] = snapshot.Offset;
            }

            return table;
        }

        #endregion

        #region Cache

        private FTable<TKey, TRecord> Retrieve(uint depth, bool createIfNotExists)
        {
            FTable<TKey, TRecord> table = depth >= Cache.Length ? null : Cache[depth];
            if (table != null)
                return table;

            ulong offset;
            if (Scheme.TryGet(depth, out offset))
                table = new FTable<TKey, TRecord>(RawFile, offset);
            else
            {
                if (!createIfNotExists)
                    return null;

                table = new FTable<TKey, TRecord>(RawFile, RawFile.Length);
                Scheme[depth] = table.Offset;
                table.CacheCapacity = CacheCapacity;
                table.BlockCapacity = BlockCapacity;
                table.KeyMap = KeyMap;

                table.Depth = depth;
                if (MaxIndex.Length < depth)
                {
                    MaxIndex = new byte[depth];
                    for (int i = 0; i < MaxIndex.Length; i++)
                        MaxIndex[i] = byte.MaxValue;
                }
            }
            table.recordPersist = RecordPersist;
            table.Cache = SharedCache;
            Packet(depth, table);

            return table;
        }

        private void Packet(uint depth, FTable<TKey, TRecord> table)
        {
            if (depth >= Cache.Length)
                Array.Resize<FTable<TKey, TRecord>>(ref Cache, (int)(depth + 1));
            Cache[depth] = table;
        }

        public void Flush()
        {
            Scheme.Flush();
            foreach (var table in Cache)
                if (table != null)
                    table.Flush();
        }

        #endregion

        #region Properties

        /// <summary>
        /// Gets the version of the table.
        /// </summary>
        public ushort Version { get; private set; }
        /// <summary>
        /// Gets the number of rows actually contained in the table.
        /// </summary>
        public ulong Count { get; private set; }

        /// <example> This sample shows how to implement custom persist.
        /// <code>
        /// //Sample record
        ///public class Tick
        ///{
        ///    public string Symbol { get; set; }
        ///    public DateTime Timestamp { get; set; }
        ///    public double Bid { get; set; }
        ///    public double Ask { get; set; }
        ///    public long BidSize { get; set; }
        ///    public long AskSize { get; set; }
        ///    public string BidExg { get; set; }
        ///    public string AskExg { get; set; }
        ///
        ///    public Tick()
        ///    {
        ///    }
        ///}
        ///
        ///[Serializable]
        ///public class TickBinaryPersist : IBinaryPersist&lt;IIndexer&lt;Tick&gt;&gt;
        ///{
        ///    private void Write(BinaryWriter writer, string value)
        ///    {
        ///        if (value == null)
        ///            writer.Write(false);
        ///        else
        ///        {
        ///            writer.Write(true);
        ///            writer.Write(value);
        ///        }
        ///    }
        ///
        ///    private string ReadString(BinaryReader reader)
        ///    {
        ///        if (!reader.ReadBoolean())
        ///            return null;
        ///        else
        ///            return reader.ReadString();
        ///    }
        ///
        ///    #region IBinaryPersist&lt;IIndexer&lt;Tick&gt;&gt; Members
        ///
        ///    public void Load(BinaryReader reader, ref IIndexer&lt;Tick&gt; data)
        ///    {
        ///        for (int i = 0; i &lt; data.Count; i++)
        ///        {
        ///            Tick tick = new Tick();
        ///
        ///            tick.Symbol = ReadString(reader);
        ///            tick.Timestamp = new DateTime(reader.ReadInt64());
        ///            tick.Bid = reader.ReadDouble();
        ///            tick.Ask = reader.ReadDouble();
        ///            tick.BidSize = reader.ReadInt64();
        ///            tick.AskSize = reader.ReadInt64();
        ///            tick.BidExg = ReadString(reader);
        ///            tick.AskExg = ReadString(reader);
        ///
        ///            data[i] = tick;
        ///        }
        ///    }
        ///
        ///    public void Store(BinaryWriter writer, IIndexer&lt;Tick&gt; data)
        ///    {
        ///        for (int i = 0; i &lt; data.Count; i++)
        ///        {
        ///            Tick tick = data[i];
        ///
        ///            Write(writer, tick.Symbol);
        ///            writer.Write(tick.Timestamp.Ticks);
        ///            writer.Write(tick.Bid);
        ///            writer.Write(tick.Ask);
        ///            writer.Write(tick.BidSize);
        ///            writer.Write(tick.AskSize);
        ///            Write(writer, tick.BidExg);
        ///            Write(writer, tick.AskExg);
        ///        }
        ///    }
        ///
        ///    #endregion
        ///}
        /// </code>
        /// </example>
        public IBinaryPersist<IIndexer<TRecord>> RecordPersist
        {
            get { return ServiceTable.RecordPersist; }
            set
            {
                if (Count > 0)
                    return;

                ServiceTable.RecordPersist = value;
            }
        }

        /// <summary>
        /// Sertain types have a predefined IKeyMap - (Byte, SByte, Boolean, Char, UInt16, UInt32, UInt64, Int16, Int32, Int64, Single, Double, String, Locator, Struct).
        /// For a custom type you need to implement IKeyMap&lt;TKey&gt;
        /// </summary>
        /// <example> This sample shows how to implement custom KeyMap.
        /// <code>
        ///[Serializable]
        ///public class UInt32KeyMap : IKeyMap&lt;UInt32&gt;
        ///{
        ///    #region IKeyMap&lt;UInt32&gt; Members
        ///
        ///    public byte[] Direct(UInt32 key)
        ///    {
        ///        return BitConverter.GetBytes(key);
        ///    }
        ///
        ///    public UInt32 Reverse(byte[] index)
        ///    {
        ///        return BitConverter.ToUInt32(index, 0);
        ///    }
        ///
        ///    #endregion
        ///}
        /// </code>
        /// </example>
        public IKeyMap<TKey> KeyMap
        {
            get { return ServiceTable.KeyMap; }
            set
            {
                if (Count > 0)
                    return;

                ServiceTable.KeyMap = value;
            }
        }

        public uint CacheCapacity
        {
            get { return ServiceTable.CacheCapacity; }
            set
            {
                if (CacheCapacity == value)
                    return;

                ServiceTable.CacheCapacity = value;

                foreach (var row in Scheme.Forward())
                {
                    FTable<TKey, TRecord> table = Retrieve(row.Key, false);
                    table.CacheCapacity = value;
                }
            }
        }

        public uint BlockCapacity
        {
            get { return ServiceTable.BlockCapacity; }
            set
            {
                if (BlockCapacity == value)
                    return;

                ServiceTable.BlockCapacity = value;

                foreach (var row in Scheme.Forward())
                {
                    FTable<TKey, TRecord> table = Retrieve(row.Key, false);
                    table.BlockCapacity = value;
                }
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <returns>null if table is empty otherwise first row.</returns>
        public Row<TKey, TRecord> FirstRow
        {
            get { return Forward().FirstOrDefault(); }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <returns>null if table is empty otherwise last row.</returns>
        public Row<TKey, TRecord> LastRow
        {
            get { return Backward().FirstOrDefault(); }
        }

        /// <summary>
        /// Gets system size of the table.
        /// </summary>
        /// <remarks>
        /// If table is modified then Size is valid only after calling Flush(), Commit() or Close().
        /// </remarks>
        public ulong SystemSize
        {
            get
            {
                ulong sum = HEADER_SIZE + Scheme.SystemSize + Scheme.DataSize;
                foreach (var row in Scheme.Forward())
                {
                    FTable<TKey, TRecord> table = Retrieve(row.Key, false);
                    sum += table.SystemSize;
                }

                return sum;
            }
        }

        /// <summary>
        /// Gets data size of the table.
        /// </summary>
        /// <remarks>
        /// If table is modified then Size is valid only after calling Flush(), Commit() or Close().
        /// </remarks>
        public ulong DataSize
        {
            get
            {
                ulong sum = 0;
                foreach (var row in Scheme.Forward())
                {
                    FTable<TKey, TRecord> table = Retrieve(row.Key, false);
                    sum += table.DataSize;
                }

                return sum;
            }
        }

        /// <summary>
        /// Gets total size of the table(SystemSize + DataSize).
        /// </summary>
        /// <remarks>
        /// If table is modified then Size is valid only after calling Flush(), Commit() or Close().
        /// </remarks>
        public ulong Size
        {
            get { return SystemSize + DataSize; }
        }

        public object SyncRoot
        {
            get { return RawFile.SyncRoot; }
        }

        #endregion

        #region ITransaction Members

        public IEnumerable<JournalItem> Journal()
        {
            foreach (var table in Cache.Where(x => x != null && x.IsModified))
            {
                foreach (JournalItem journalItem in table.Journal())
                    yield return journalItem;
            }

            if (Scheme.IsModified)
            {
                foreach (JournalItem journalItem in Scheme.Journal())
                    yield return journalItem;
            }

            if (ServiceTable.IsModified)
            {
                foreach (JournalItem journalItem in ServiceTable.Journal())
                    yield return journalItem;
            }

            yield return new JournalItem(RawFile.Handle, Offset, BitConverter.GetBytes(ActiveHeader));
        }

        /// <summary>
        /// Apply (commit) the transaction.
        /// </summary>
        public void Commit()
        {
            //FTable.Offset -> FTable.ActiveHeader
            FTable<ulong, byte> journalTable = new FTable<ulong, byte>(RawFile, Offset + HEADER_SIZE);
            journalTable.KeyMap = new FixedUInt64KeyMap();

            if (Scheme.IsModified)
                journalTable[Scheme.Offset] = Scheme.ActiveHeader;

            if (ServiceTable.IsModified)
                journalTable[ServiceTable.Offset] = ServiceTable.ActiveHeader;

            foreach (var table in Cache.Where(x => x != null && x.IsModified))
                journalTable[table.Offset] = table.ActiveHeader;

            if (journalTable.Count == 0) //there is no modifications
                return;

            journalTable[this.Offset] = this.ActiveHeader;
            journalTable.Commit();

            foreach (var table in Cache.Where(x => x != null && x.IsModified))
                table.Commit();

            Scheme.Commit();
            ServiceTable.Commit();

            byte activeHeader = (byte)(ActiveHeader == 0 ? 1 : 0);
            SaveHeader(activeHeader);

            RawFile.Flush();

            BinaryFile binaryFile = new BinaryFile(RawFile);
            binaryFile.Position = Offset;
            binaryFile.Write(activeHeader);

            RawFile.Flush();

            journalTable.Clear();
            journalTable.Commit();

            ActiveHeader = activeHeader;
        }

        /// <summary>
        /// Rolls back (abort) the transaction.
        /// </summary>
        public void Rollback()
        {
            BinaryFile binaryFile = new BinaryFile(RawFile);
            //FTable.Offset -> FTable.ActiveHeader
            FTable<ulong, byte> journalTable = new FTable<ulong, byte>(RawFile, Offset + HEADER_SIZE);
            journalTable.KeyMap = new FixedUInt64KeyMap();
            foreach (var row in journalTable)
            {
                binaryFile.Position = row.Key;
                binaryFile.Write(row.Record);
            }
            journalTable.Clear();
            journalTable.Commit();

            Scheme = new FTable<uint, ulong>(RawFile, journalTable.Offset + FTable<ulong, byte>.HEADER_SIZE);
            ServiceTable = new FTable<TKey, TRecord>(RawFile, Scheme.Offset + FTable<uint, ulong>.HEADER_SIZE);

            Cache = new FTable<TKey, TRecord>[Cache.Length];
            SharedCache.Clear();

            LoadHeader(ActiveHeader);
        }

        #endregion

        #region Save/Load

        private void SaveHeader(byte activeHeader)
        {
            BinaryFile binaryFile = new BinaryFile(RawFile);
            binaryFile.Position = Offset + sizeof(byte) + activeHeader * HALF_HEADER_SIZE;
            binaryFile.Write(Version);
            binaryFile.Write(Count);
            binaryFile.Write(MaxIndex.Length);
        }

        private void LoadHeader(byte activeHeader)
        {
            BinaryFile binaryFile = new BinaryFile(RawFile);
            binaryFile.Position = Offset + sizeof(byte) + activeHeader * HALF_HEADER_SIZE;
            Version = binaryFile.ReadUInt16();
            Count = binaryFile.ReadUInt64();
            MaxIndex = new byte[binaryFile.ReadInt32()];
            for (int i = 0; i < MaxIndex.Length; i++)
                MaxIndex[i] = byte.MaxValue;
        }

        #endregion
    }
}
